package cn.tang.tframe.mq.kafka.interceptor;

import cn.tang.tframe.common.annotation.BeanOrder;
import cn.tang.tframe.common.utils.lang.StringUtils;
import cn.tang.tframe.mq.IQtMessager;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

/**
 * Created by tanghc on 17/2/4.
 */
@BeanOrder(order = -1)
public class QtMonitorKafkaConsumerInterceptor implements ConsumerInterceptor<IQtMessager,IQtMessager> {

	@Override
	public ConsumerRecords<IQtMessager, IQtMessager> onConsume(ConsumerRecords<IQtMessager, IQtMessager> records) {
		if(null != records){
			for(ConsumerRecord<IQtMessager, IQtMessager> record: records){
				IQtMessager msg = record.value();
				if(null != msg){
					System.out.println("=====test monitor kafka==="+ StringUtils.join(msg.getAttach().entrySet()));
				}
			}
		}
		return records;
	}

	@Override
	public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {

	}

	@Override
	public void close() {

	}

	@Override
	public void configure(Map<String, ?> configs) {

	}
}
